package com.mindata.blockchain.socket.raft.server;

import com.mindata.blockchain.socket.raft.core.proto.RaftMessage;
import com.github.wenweihu86.rpc.server.RPCServer;
import com.mindata.blockchain.common.RaftConf;
import com.mindata.blockchain.common.SystemConfig;
import com.mindata.blockchain.core.manager.PermissionManager;
import com.mindata.blockchain.socket.raft.core.RaftNode;
import com.mindata.blockchain.socket.raft.core.RaftOptions;
import com.mindata.blockchain.socket.raft.core.service.RaftClientService;
import com.mindata.blockchain.socket.raft.core.service.RaftConsensusService;
import com.mindata.blockchain.socket.raft.core.service.impl.RaftClientServiceImpl;
import com.mindata.blockchain.socket.raft.core.service.impl.RaftConsensusServiceImpl;
import com.mindata.blockchain.socket.raft.utils.RaftUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

/**
 * @Author:wenhui
 * @Description:raft的server启动器
 * @Date:Create in 2018-7-25 16:17
 **/
@Component
@Slf4j(topic = "RaftServerStarter")
public class RaftServerStarter {

    @Resource
    private PermissionManager permissionManager;
    @Resource
    private SystemConfig systemConfig;

    @PostConstruct
    public void start() {
        //TODO 联盟节点动态变更
        String clusterPeers = permissionManager.getRaftClusterPeers();
        if (StringUtils.isBlank(clusterPeers)) {
            log.warn("无联盟节点配置");
            System.exit(0);
        }
        log.info("raft cluster peers:{}", clusterPeers);

        List<RaftMessage.Server> serverList = RaftUtils.parseServers(clusterPeers);
        // local server
        RaftMessage.Server localServer = RaftUtils.parseServer(systemConfig.getRaftLocalServer());

        // 初始化RPCServer
        RPCServer server = new RPCServer(localServer.getEndPoint().getPort());
        // 设置Raft选项，比如：
        // just for test snapshot
        RaftOptions raftOptions = new RaftOptions();
        raftOptions.setDataDir(RaftConf.logDataPath);
        raftOptions.setSnapshotMinLogSize(10 * 1024);
        raftOptions.setSnapshotPeriodSeconds(30);
        raftOptions.setMaxSegmentFileSize(1024 * 1024);
        // 应用状态机
        RaftStateMachine stateMachine = RaftStateMachineContainer.getInstance(raftOptions.getDataDir());
        // 初始化RaftNode
        RaftNode raftNode = new RaftNode(raftOptions, serverList, localServer, stateMachine);
        // 注册Raft节点之间相互调用的服务
        RaftConsensusService raftConsensusService = new RaftConsensusServiceImpl(raftNode);
        server.registerService(raftConsensusService);
        // 注册给Client调用的Raft服务
        RaftClientService raftClientService = new RaftClientServiceImpl(raftNode);
        server.registerService(raftClientService);
        // 注册应用自己提供的服务
        RaftServerService raftServerService = new RaftServerServiceImpl(raftNode, stateMachine);
        server.registerService(raftServerService);
        // 启动RPCServer，初始化Raft节点
        server.start();
        raftNode.init();
    }
}
